package com.chenxin.transaction.message;

import com.alibaba.fastjson.JSONObject;
import com.chenxin.transaction.dao.ZhangsanAccountInfoDao;
import com.chenxin.transaction.entity.AccountChangeEvent;
import com.chenxin.transaction.service.AccountInfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.transaction.annotation.Transactional;

/**
 * @Description:
 * @title: ProducerListener
 * @Author Star_Chen
 * @Date: 2022/6/1 9:33
 * @Version 1.0
 */
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
public class ProducerListener implements RocketMQLocalTransactionListener {

    @Autowired
    private AccountInfoService accountInfoService;

    @Autowired
    private ZhangsanAccountInfoDao zhangsanAccountInfoDao;

    /**
     * @param message :
     * @Date: 2022/6/1 9:34
     * @Description: 事务消息发送后的回调方法，当消息发送给mq成功，此方法被mq回调
     * 如果生产者在发完消息后，突然宕机了，消息就死在了队列里，永远都是待消费状态，所以mq要进行回查此事务，进行扣款操作
     * @Return: * @return : org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     */
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        try {
            //带着消息来解析
            String messageString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            String accountChange = jsonObject.getString("accountChange");
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChange, AccountChangeEvent.class);

            //模拟这里出现了RuntimeException
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.info("事务执行失败！！");
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }

    /**
     * @Date: 2022/6/1 10:05
     * @Description: 检查事务执行的状态，事务状态回查，查询是否扣减金额
     * 如果生产者在扣款后，突然宕机了或者超时，消息就死在了队列里，永远都是待消费状态，所以mq要进行回查此事务
     * @param message :
     * @Return:  * @return : org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析message，转成AccountChangeEvent
        String messageString = new String((byte[]) message.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String accountChangeString = jsonObject.getString("accountChange");
        //将accountChange（json）转成AccountChangeEvent
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //事务id
        String txNo = accountChangeEvent.getTxNo();
        int existTx = zhangsanAccountInfoDao.isExistTx(txNo);
        if(existTx>0){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            // unknown表示继续回调重试
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
